import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 此类的作用是将自定义的mapper和reducer的job任务提交到hadoop集群上，分发执行的客户端
 * 
 * @author knight
 *
 */
public class JobSubmit {
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		Job wordCountJob = Job.getInstance(conf);

		// 重要：要设置本job所在的jar包
		wordCountJob.setJarByClass(JobSubmit.class);

		// 设置job用的mapper逻辑是哪个类
		wordCountJob.setMapperClass(WordCountMapper.class);
		// 设置job用的reducer逻辑是哪个类
		wordCountJob.setReducerClass(WordCountReducer.class);

		// 设置mapper输出的KV的类型
		wordCountJob.setMapOutputKeyClass(Text.class);
		wordCountJob.setMapOutputValueClass(IntWritable.class);
		// 设置最终输出的KV数据类型
		wordCountJob.setOutputKeyClass(Text.class);
		wordCountJob.setOutputValueClass(IntWritable.class);

		// 设置要处理的文本数据的存放路径
		FileInputFormat.setInputPaths(wordCountJob, "hdfs://hdp01:9000/mapreduce/wordcount/data");
		FileOutputFormat.setOutputPath(wordCountJob, new Path("hdfs://hdp01:9000/mapreduce/wordcount/result"));

		// 提交job给hadoop集群
		wordCountJob.waitForCompletion(true);
	}
}
